package org.example.kafka.sendtype;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.example.kafka.common.CommonConstant;
import org.example.kafka.common.KafkaConstants;

/**
 * 异步发送消息
 */
public class KafkaAsynProducer {
    private static KafkaProducer<String,String> producer = null;

    public static void main(String[] args) {
        /*消息生产者*/
        producer = new KafkaProducer<String, String>(
                KafkaConstants.producerConfig(StringSerializer.class,
                        StringSerializer.class));
        // 待发送的消息实例
        ProducerRecord<String,String> record;
        try{
            record = new ProducerRecord<String,String>(CommonConstant.HELLO_TOPIC,"kafkaHello","hero");
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata,
                                         Exception exception) {
                    if(null!=exception){
                        exception.printStackTrace();
                    }
                    if(null!=metadata){
                        System.out.println("offset:"+metadata.offset()+"-----------"
                                +"partition:"+metadata.partition());
                    }
                }
            });
        }finally {
            producer.close();
        }
    }
}
